subprocess.Popen(
    args,                # 命令列表（推荐）: ['ls', '-l']；或字符串（需 shell=True）
    shell=False,         # ⚠️ 避免对用户输入设为 True（安全风险）
    stdout=PIPE,         # 捕获输出: PIPE / DEVNULL / 文件对象
    stderr=PIPE,         # 错误流处理（可设为 STDOUT 合并）
    stdin=PIPE,          # 传递输入
    cwd='/path',         # 指定工作目录
    env={'KEY':'VAL'},   # 自定义环境变量（需 copy 原环境）
    text=True,           # Python 3.7+：自动处理字符串（替代 universal_newlines）
    encoding='utf-8'     # 指定编码（配合 text 使用）
)

 

代码举例:

import subprocess
import time
from subprocess import Popen

cmd_str = 'ps aux|grep sys'
print('exec...', cmd_str)
with Popen(cmd_str, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) as f1:
    while True:
        buff = f1.stdout.readline()
        ret = Popen.poll(f1)
        line = None
        if buff:
            line = buff.decode(encoding='utf-8').strip('\r\n')
            print('...', line)
        if ret is not None:
            print('\r\n...exec finish.')
            break

time.sleep(1)

 

Python subprocess.Popen 使用指南

subprocess.Popen 是 Python 中用于异步创建和精细控制子进程的核心类，适用于需要实时交互、流式处理或长时间运行的外部程序调用场景。

---

🔑 核心要点速览

项目	说明
适用场景	需要实时读写、异步控制、流式处理的复杂场景
关键方法	communicate()（推荐）、wait()、poll()、terminate()
安全原则	优先使用列表参数 + shell=False（防注入）
资源管理	务必调用 communicate() 或 wait() 避免僵尸进程
替代方案	简单同步调用 → 优先用 subprocess.run()（Python 3.5+）

---

📌 常用参数说明

subprocess.Popen(
    args,                # 命令列表（推荐）: ['ls', '-l']；或字符串（需 shell=True）
    shell=False,         # ⚠️ 避免对用户输入设为 True（安全风险）
    stdout=PIPE,         # 捕获输出: PIPE / DEVNULL / 文件对象
    stderr=PIPE,         # 错误流处理（可设为 STDOUT 合并）
    stdin=PIPE,          # 传递输入
    cwd='/path',         # 指定工作目录
    env={'KEY':'VAL'},   # 自定义环境变量（需 copy 原环境）
    text=True,           # Python 3.7+：自动处理字符串（替代 universal_newlines）
    encoding='utf-8'     # 指定编码（配合 text 使用）
)

---

💡 实用代码示例

1️⃣ 捕获输出与错误（推荐方式）

import subprocess

proc = subprocess.Popen(
    ['python', '--version'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True  # 直接返回字符串（Python 3.7+）
)
stdout, stderr = proc.communicate()  # 安全读取，自动等待结束
print("输出:", stdout.strip() or stderr.strip())
print("返回码:", proc.returncode)

2️⃣ 向子进程传递输入

proc = subprocess.Popen(
    ['grep', 'hello'],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    text=True
)
output, _ = proc.communicate(input="hello world\nhi there")
print(output)  # 输出: hello world

3️⃣ 超时控制与终止

try:
    proc = subprocess.Popen(['sleep', '10'], stdout=subprocess.PIPE)
    stdout, _ = proc.communicate(timeout=3)  # 3秒超时
except subprocess.TimeoutExpired:
    proc.kill()  # 强制终止
    proc.communicate()  # 清理残留输出
    print("进程已超时终止")

4️⃣ 合并错误流 + 设置环境

import os
env = os.environ.copy()
env['CUSTOM_VAR'] = 'test'

proc = subprocess.Popen(
    ['./script.sh'],
    cwd='/project',
    env=env,
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,  # 合并 stderr 到 stdout
    text=True
)
output, _ = proc.communicate()

---

⚠️ 重要注意事项

1. 避免死锁：切勿直接 proc.stdout.read()，务必用 communicate() 读写
2. 安全第一：用户输入拼接命令时，禁用 shell=True，使用参数列表

   # ❌ 危险（命令注入风险）
   subprocess.Popen(f"ls {user_input}", shell=True)
   
   # ✅ 安全
   subprocess.Popen(['ls', user_input])
   

3. 资源清理：即使异常也要确保进程结束（可用 try/finally 或上下文管理器）
4. 跨平台差异：Windows 命令（如 dir）与 Linux（ls）不同，注意兼容性
5. 编码处理：若未设 text=True，输出为 bytes，需手动 .decode('utf-8')

---

🔄 Popen vs run()

特性	Popen	subprocess.run()
控制粒度	高（实时交互、异步）	低（同步阻塞）
使用复杂度	较高	简单（推荐日常使用）
适用场景	流式处理、长时间任务	一次性命令执行
返回结果	Popen 对象	CompletedProcess 对象

✅ 简单任务首选：

result = subprocess.run(['ls', '-l'], capture_output=True, text=True)
print(result.stdout)

---

掌握 Popen 可实现对子进程的精准控制，但务必注意安全性、资源管理和编码问题。对于大多数同步场景，subprocess.run() 更简洁安全！